package com.syg.kafka03;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @author syg
 */
public class Producer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 配置参数
        Properties properties = new Properties();
        // bootstrap.servers：broker的地址清单
        properties.put("bootstrap.servers", "localhost:9092");
        // key.serializer：消息的键的序列化器
        properties.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        // value.serializer：消息的值的序列化器
        properties.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        // 2. 根据参数创建KafkaProducer实例（生产者）
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        // 3. 创建ProducerRecord实例（消息）
        ProducerRecord<String, String> record = new ProducerRecord<>("topic-demo", "hello kafka");
        // 4. 发送消息  异步调用
        producer.send(record, new Callback());
        // 5. 关闭生产者示例
        producer.close();
    }
}


/**
 *  这样是异步调用？还是同步调用？
 *  Future<RecordMetadata> sendFuture = producer.send(record, new Callback());
 *  // send方法的返回值是一个Future对象，当调用其get方法时将阻塞等待Kafka的响应
 *  RecordMetadata recordMetadata = sendFuture.get();
 */
class Callback implements org.apache.kafka.clients.producer.Callback{

    /**
     * onCompletion有两个参数，其类型分别是RecordMetadata和Exception。当消息发送成功时，recordMetadata为非null，而e将为null。当消息发送失败时，则反之。
     * @param recordMetadata
     * @param e
     */
    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        if (e != null) {
            e.printStackTrace();
        } else {
            System.out.println(recordMetadata.topic() + "-" + recordMetadata.partition() + ":" + recordMetadata.offset());
        }
    }
}
